package com.rabbitmq.comfig;

import com.rabbitmq.rabbit.OrderMessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.backoff.NoBackOffPolicy;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author: TongRui乀
 * @Date: 2019/8/2 10:12
 * @description：
 */
@Configuration
public class RabbitmqConfig {


    private static final Logger log= LoggerFactory.getLogger(RabbitmqConfig.class);

    @Autowired
    private Environment env;

    @Autowired
    private CachingConnectionFactory connectionFactory;

    @Autowired
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    @Bean
    public RabbitAdmin rabbitAdmin(){
        return new RabbitAdmin(connectionFactory);
    }

    /**
     * 单一消费者
     * @return
     */
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainer(){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        factory.setTxSize(1);
        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
        return factory;
    }

    /**
     * 多个消费者
     * @return
     */
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer(){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory,connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        factory.setConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.concurrency",int.class));
        factory.setMaxConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.max-concurrency",int.class));
        factory.setPrefetchCount(env.getProperty("spring.rabbitmq.listener.prefetch",int.class));
        factory.setRetryTemplate(createRetry());
        return factory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(){
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> log.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause));
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message));
        return rabbitTemplate;
    }

    /****************** Direct交换机模式 *******************/

    @Bean
    public Queue logQueue(){
        return new Queue(env.getProperty("log.queue.name"), true);
    }


    @Bean
    public Queue logMultiQueue(){

        HashMap<String, Object> map = new HashMap<>();

        map.put("x-dead-letter-exchange", "dead.exchange");
        map.put("x-dead-letter-routing-key", env.getProperty("log.multi.consumer.routing.key.name"));

        return QueueBuilder.durable(env.getProperty("log.queue.multi.consumer.name")).withArguments(map).build();
    }

    /**
     *  死信队列
     * @return
     */
    @Bean
    public Queue deadQueue(){

        return QueueBuilder.durable(env.getProperty("dead.queue.name")).build();
    }

    @Bean
    public DirectExchange logExchange(){
        return new DirectExchange(env.getProperty("log.exchange.name"), true, false);
    }

    @Bean
    public TopicExchange deadExchange(){
        return new TopicExchange("dead.exchange", true, false);
    }

    @Bean
    public Binding logBinding(){
        return BindingBuilder.bind(logQueue()).to(logExchange()).with(env.getProperty("log.routing.key.name"));
    }

    @Bean
    public Binding logMultiBingding(){
        return BindingBuilder.bind(logMultiQueue()).to(logExchange()).with(env.getProperty("log.multi.consumer.routing.key.name"));
    }

    /**
     *  死信队列绑定到交换机
     * @return
     */
    @Bean
    public Binding deadBinding(){
        return BindingBuilder.bind(deadQueue()).to(deadExchange()).with(env.getProperty("log.multi.consumer.routing.key.name"));
    }

    /***************    延时消息     *************/

    /**
     *  延迟队列
     * @return
     */
    @Bean
    public Queue delayQueue(){
        return QueueBuilder.durable(env.getProperty("order.delay.queue.name")).build();
    }

    /**
     *  延时交换机
     * @return
     */
    @Bean
    public CustomExchange delayExchange(){
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(env.getProperty("order.delay.exchange.name"), "x-delayed-message", true, false, args);
    }

    /**
     *  绑定延迟队列
     * @return
     */
    @Bean
    public Binding delayBinding(){
        return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(env.getProperty("order.delay.queue.key")).noargs();
    }

    /***************    延时消息     *************/



    private RetryTemplate createRetry() {
        RetryTemplate retryTemplate = new RetryTemplate();
        retryTemplate.registerListener(new RetryListener() {
            @Override
            public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
                // 第一次重试调用
                log.error("第一次重试调用");
                return false;
            }

            @Override
            public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {
                // 最后一次重试会调用
                log.error("最后一次重试会调用");
            }

            @Override
            public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {
                // 每次重试失败后都会调用
                log.error("每次重试失败后都会调用");
            }
        });
        retryTemplate.setRetryPolicy(new SimpleRetryPolicy(5));
        retryTemplate.setBackOffPolicy(new NoBackOffPolicy());
        return retryTemplate;
    }

}
